package cn.kgc.gmall.app.dwd;

import cn.kgc.gmall.utils.MyKafkaUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;

/**
 *  读取(消费)kafka中ods层的数据
 *  将json字符串转化为json对象
 *  对是否是新用户做一个重新处理
 *  使用侧输出流  分别写到不同的主题中
 */
public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        // 1.创建好运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 2.设置并行度
        env.setParallelism(4);
        // 3.激活检查点 每5秒开启一次检查点  精准一次
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        // 4.设置检查点重复 次数 一分钟之内如果还没有完成检测点的创建操作那就抛弃
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        // 4.1 如果flink程序关闭是否删除检查点
        env.getCheckpointConfig()
                .enableExternalizedCheckpoints(
                        CheckpointConfig
                                .ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
                );
        // 5.状态后端 保存到hdfs
        env.setStateBackend(
                new FsStateBackend("hdfs://hadoop102:8020/gmall/flink/checkpoint")
        );
        // 5.1 模拟用户 设置用户
        System.setProperty("HADOOP_USER_NAME","atkgc");
        String groupId = "base_log_group";
        String topic  = "ods_base_log";
        // 6.获取kafka连接
        DataStreamSource<String> baseLogDS = env.addSource(MyKafkaUtils.getKafkaConsumerSource(topic,groupId));
        // 7.转换为json对象  String -> jsonObject
        SingleOutputStreamOperator<JSONObject> jsonObjectDS = baseLogDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String s) throws Exception {
                return JSON.parseObject(s);
            }
        });
        // 函数式编程
        // baseLogDs.map(x -> JSON.parseObject(x));
        //jsonObjectDS.print();

        // 8.根据mid 进行分组
        KeyedStream<JSONObject, String> keyByDS = jsonObjectDS.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getJSONObject("common").getString("mid");
            }
        });
        // 8.1 处理新用户修复   每条key只出现一次有可能是新用户  如果出现多次判断是不是同一天
        SingleOutputStreamOperator<JSONObject> mapDS = keyByDS.map(new RichMapFunction<JSONObject, JSONObject>() {
            // 状态变量 存储当前key的mid
            private ValueState<String> dateState = null;
            // 状态变量 存储当前数据的时间
            private SimpleDateFormat dateFormat = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                dateState = getRuntimeContext().getState(
                        new ValueStateDescriptor<String>("dateState", String.class)
                );
                dateFormat = new SimpleDateFormat("yyyy-MM-dd");

            }

            @Override
            public JSONObject map(JSONObject jsonObject) throws Exception {

                // 获取is_new
                String isNew = jsonObject.getJSONObject("common").getString("is_new");
                // 获取状态变量中的值
                String newMidDate = dateState.value();
                // 获取当前数据的时间戳  转换成年月日
                String ts = dateFormat.format(jsonObject.getLong("ts"));
                // 判断用户可能不是新用户
                if (isNew.equals("1")) {

                    // 判断状态变量中是否有值
                    if (newMidDate != null && newMidDate.length() > 0) {  // 说明不是新用户
                        // 判断是否是同一天  如果不是同一天的操作说明是老用户 将1修复成0
                        if (!dateState.value().equals(ts)) {
                            isNew = "0";
                            jsonObject.getJSONObject("common").put("is_new", isNew); // 完成修复操作
                        }
                    } else {
                        // 有可能是新用户
                        dateState.update(dateFormat.format(jsonObject.getLong("ts")));
                    }
                }
                return jsonObject;
            }
        });

        // 9.使用侧输出流将数据按照不同的主题输出到Kafka中的主题中
        // 曝光数据
        OutputTag<String> displaysStream = new OutputTag<String>("displaysStream"){};
        // 启动数据
        OutputTag<String> startStream = new OutputTag<String>("startStream"){};
        // 处理函数
        SingleOutputStreamOperator<String> processDS = mapDS.process(new ProcessFunction<JSONObject, String>() {

            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, String>.Context context, Collector<String> collector) throws Exception {
                // 判断是否是启动数据
                JSONObject start = jsonObject.getJSONObject("start");
                if (start != null && start.size() > 0) {
                    // 标记写出到start侧输出流里面
                    context.output(startStream, jsonObject.toJSONString());
                } else {
                    // 页面数据正常输出
                    collector.collect(jsonObject.toJSONString());
                    // 曝光数据
                    JSONArray displays = jsonObject.getJSONArray("displays");
                    if (displays != null && displays.size() > 0) {
                        // 获取数据的时间
                        String ts = jsonObject.getString("ts");
                        for (int i = 0; i < displays.size(); i++) {
                            String pageId = jsonObject.getJSONObject("page").getString("page_id");
                            // 获取数组中每个json对象
                            JSONObject displaysObject = displays.getJSONObject(i);
                            // 时间
                            displaysObject.put("ts", ts);
                            // 页面
                            displaysObject.put("page_id", pageId);
                            context.output(displaysStream, displaysObject.toString());
                        }
                    }
                }


            }
        });

        // 打印每条侧输出流的数据
        /*processDS.getSideOutput(displaysStream).print("###");
        processDS.getSideOutput(startStream).print("***");
        processDS.print(">>>");*/

        // 将侧输出流中的数据写出到kafka

        DataStream<String> displaysOutput = processDS.getSideOutput(displaysStream);
        DataStream<String> startOutput = processDS.getSideOutput(startStream);
        // 曝光数据
        displaysOutput.addSink(MyKafkaUtils.getKafkaProducerSink("dwd_topic_displays"));
        // 启动数据
        startOutput.addSink(MyKafkaUtils.getKafkaProducerSink("dwd_topic_start"));
        // 页面数据
        processDS.addSink(MyKafkaUtils.getKafkaProducerSink("dwd_topic_page"));

        env.execute();

    }
}
